#include "mqconnectionpool.h"
#include <QDebug>
#include <QThread>
#include <QCoreApplication>
#include <amqp_tcp_socket.h>

Q_LOGGING_CATEGORY(connectionpool, "mq.connectionpool")

/**
 * @brief 构造函数
 * @param config 连接配置
 * @param poolSize 连接池大小
 * @param parent 父对象
 */
MQConnectionPool::MQConnectionPool(const MQConnectionConfig& config, int poolSize, QObject* parent)
    : QObject(parent)
    , m_config(config)
    , m_maxConnections(poolSize)
    , m_minConnections(qMax(1, poolSize / 4))  // 最小连接数为最大连接数的1/4
    , m_connectionTimeoutMs(5000)
    , m_maxIdleTimeMs(300000)  // 5分钟
    , m_initialized(false)
    , m_shuttingDown(false)
{
    // 初始化定时器
    m_cleanupTimer = std::make_unique<QTimer>(this);
    
    // 连接定时器信号
    connect(m_cleanupTimer.get(), &QTimer::timeout, this, &MQConnectionPool::cleanupExpiredConnections);
    
    // 初始化统计信息
    memset(&m_statistics, 0, sizeof(m_statistics));
    
    qCDebug(connectionpool) << "连接池创建完成，最大连接数:" << m_maxConnections;
}

/**
 * @brief 析构函数
 */
MQConnectionPool::~MQConnectionPool()
{
    cleanup();
    qCDebug(connectionpool) << "连接池已销毁";
}

/**
 * @brief 初始化连接池
 * @param config 连接配置
 * @return 是否初始化成功
 */
bool MQConnectionPool::initialize(const MQConnectionConfig& config)
{
    QMutexLocker locker(&m_connectionsMutex);
    
    if (m_initialized) {
        qCWarning(connectionpool) << "连接池已经初始化";
        return true;
    }
    
    m_config = config;
    m_shuttingDown = false;
    
    // 预创建最小数量的连接
    // 在 initialize 方法中创建连接的部分
    for (int i = 0; i < m_minConnections; ++i) {
        amqp_connection_state_t conn = createConnection();
        if (conn) {
            auto connInfo = std::make_shared<MQConnectionInfo>();
            connInfo->setConnection(conn);  // 使用安全的设置方法
            connInfo->inUse = false;
            connInfo->createdTime = QDateTime::currentDateTime();
            connInfo->lastUsed = QDateTime::currentDateTime();
            
            m_connections.append(connInfo);
            qCDebug(connectionpool) << "预创建连接" << i + 1 << "成功";
        } else {
            qCWarning(connectionpool) << "预创建连接" << i + 1 << "失败";
        }
    }
    
    // 启动定时器
    m_cleanupTimer->start(m_maxIdleTimeMs / 2);  // 清理定时器频率为空闲时间的一半
    
    m_initialized = true;
    updateStatistics();
    
    qCInfo(connectionpool) << "连接池初始化完成，预创建连接数:" << m_connections.size();
    return true;
}

/**
 * @brief 获取连接
 * @param timeoutMs 超时时间（毫秒），-1表示无限等待
 * @return 连接状态指针，nullptr表示获取失败
 */
amqp_connection_state_t MQConnectionPool::getConnection(int timeoutMs)
{
    QDateTime startTime = QDateTime::currentDateTime();
    QMutexLocker locker(&m_connectionsMutex);
    
    if (!m_initialized || m_shuttingDown) {
        qCWarning(connectionpool) << "连接池未初始化或正在关闭";
        return nullptr;
    }
    
    // 查找可用的空闲连接
    for (auto& connInfo : m_connections) {
        if (!connInfo->inUse) {
            connInfo->inUse = true;
            connInfo->lastUsed = QDateTime::currentDateTime();
            
            // 记录等待时间
            qint64 waitTime = startTime.msecsTo(QDateTime::currentDateTime());
            QMutexLocker statsLocker(&m_statisticsMutex);
            m_waitTimes.enqueue(QDateTime::currentDateTime());
            if (m_waitTimes.size() > 100) {  // 保持最近100次的等待时间
                m_waitTimes.dequeue();
            }
            
            updateStatistics();
            //GJ 正常情况，暂时不打印
//            qCDebug(connectionpool) << "获取到空闲连接，等待时间:" << waitTime << "ms";
            return connInfo->connection.get();
        }
    }
    
    // 如果没有空闲连接且未达到最大连接数，创建新连接
    if (m_connections.size() < m_maxConnections) {
        amqp_connection_state_t conn = createConnection();
        if (conn) {
            auto connInfo = std::make_shared<MQConnectionInfo>();
            connInfo->connection.reset(conn);
            connInfo->inUse = true;
            connInfo->createdTime = QDateTime::currentDateTime();
            connInfo->lastUsed = QDateTime::currentDateTime();

            
            m_connections.append(connInfo);
            updateStatistics();
            
            qCDebug(connectionpool) << "创建新连接成功，当前连接数:" << m_connections.size();
            return connInfo->connection.get();
        } else {
            emit connectionCreationFailed("创建新连接失败");
            qCWarning(connectionpool) << "创建新连接失败";
        }
    }
    
    // 等待连接可用
    if (timeoutMs > 0) {
        qCDebug(connectionpool) << "等待连接可用，超时时间:" << timeoutMs << "ms";
        if (m_connectionAvailable.wait(&m_connectionsMutex, timeoutMs)) {
            // 递归调用，重新尝试获取连接
            locker.unlock();
            return getConnection(0);  // 不再等待，直接尝试
        } else {
            qCWarning(connectionpool) << "等待连接超时";
        }
    } else if (timeoutMs == -1) {
        // 无限等待
        qCDebug(connectionpool) << "无限等待连接可用";
        m_connectionAvailable.wait(&m_connectionsMutex);
        locker.unlock();
        return getConnection(0);  // 不再等待，直接尝试
    }
    
    qCWarning(connectionpool) << "获取连接失败";
    return nullptr;
}

/**
 * @brief 归还连接
 * @param conn 要归还的连接
 * @param hasError 连接是否有错误
 */
void MQConnectionPool::returnConnection(amqp_connection_state_t conn, bool hasError)
{
    if (!conn) {
        qCWarning(connectionpool) << "尝试归还空连接";
        return;
    }
    
    QMutexLocker locker(&m_connectionsMutex);
    
    auto connInfo = findConnectionInfo(conn);
    if (!connInfo) {
        qCWarning(connectionpool) << "归还的连接不属于此连接池";
        return;
    }
    
    connInfo->inUse = false;
    connInfo->lastUsed = QDateTime::currentDateTime();
    
    if (hasError) {
        connInfo->errorCount++;
        qCWarning(connectionpool) << "连接归还时发生错误，错误计数:" << connInfo->errorCount;
    } else {
        // 重置错误计数
        connInfo->errorCount = 0;
    }
    
    updateStatistics();
    
    // 通知等待的线程
    m_connectionAvailable.wakeOne();

    //GJ  无错的时候，不打印
//    qCDebug(connectionpool) << "连接已归还，是否有错误:" << hasError;
    if(hasError)
       qCDebug(connectionpool) << "连接已归还，是否有错误:" << hasError;
}

/**
 * @brief 获取连接池状态信息
 * @return 状态信息字符串
 */
QString MQConnectionPool::getStatus() const
{
    QMutexLocker locker(&m_connectionsMutex);
    
    int total = m_connections.size();
    int active = 0;
    int idle = 0;
    
    for (const auto& connInfo : m_connections) {
        if (connInfo->inUse) {
            active++;
        } else {
            idle++;
        }
    }
    
    return QString("连接池状态 - 总数:%1, 活跃:%2, 空闲:%3, 最大:%4")
           .arg(total).arg(active).arg(idle).arg(m_maxConnections);
}

/**
 * @brief 获取连接池统计信息
 * @return 统计信息结构体
 */
MQConnectionPool::PoolStatistics MQConnectionPool::getStatistics() const
{
    QMutexLocker locker(&m_statisticsMutex);
    return m_statistics;
}

/**
 * @brief 设置连接池配置
 * @param config 新的连接配置
 */
void MQConnectionPool::updateConfig(const MQConnectionConfig& config)
{
    QMutexLocker locker(&m_connectionsMutex);
    m_config = config;
    qCInfo(connectionpool) << "连接池配置已更新";
}

/**
 * @brief 清理连接池
 */
void MQConnectionPool::cleanup()
{
    QMutexLocker locker(&m_connectionsMutex);
    
    m_shuttingDown = true;
    
    // 停止定时器并断开信号连接
    if (m_cleanupTimer) {
        m_cleanupTimer->stop();
        m_cleanupTimer->disconnect();
    }
    
    // 安全关闭所有连接
    for (auto& connInfo : m_connections) {
        if (connInfo) {
            connInfo->safeReleaseConnection();
        }
    }
    
    m_connections.clear();
    m_initialized = false;
    
    // 唤醒所有等待的线程
    m_connectionAvailable.wakeAll();
    
    qCInfo(connectionpool) << "连接池清理完成";
}





/**
 * @brief 清理过期连接
 */
void MQConnectionPool::cleanupExpiredConnections()
{
    QMutexLocker locker(&m_connectionsMutex);
    
    if (m_shuttingDown) {
        return;
    }
    
    QDateTime now = QDateTime::currentDateTime();
    int removedCount = 0;
    
    // 从后往前遍历，方便删除
    for (int i = m_connections.size() - 1; i >= 0; --i) {
        auto& connInfo = m_connections[i];
        
        // 只清理空闲连接
        if (!connInfo->inUse) {
            qint64 idleTime = connInfo->lastUsed.msecsTo(now);
            
            // 如果空闲时间超过最大空闲时间，且连接数大于最小连接数
            if (idleTime > m_maxIdleTimeMs && m_connections.size() > m_minConnections) {
                closeConnection(connInfo->connection.get());
                m_connections.removeAt(i);
                removedCount++;
                qCDebug(connectionpool) << "清理过期连接，空闲时间:" << idleTime << "ms";
            }
        }
    }
    
    if (removedCount > 0) {
        updateStatistics();
        qCInfo(connectionpool) << "清理过期连接完成，移除连接数:" << removedCount;
    }
}

/**
 * @brief 创建新连接
 * @return 连接状态指针，nullptr表示创建失败
 */
amqp_connection_state_t MQConnectionPool::createConnection()
{
    amqp_connection_state_t conn = amqp_new_connection();
    if (!conn) {
        qCWarning(connectionpool) << "创建AMQP连接状态失败";
        return nullptr;
    }
    
    amqp_socket_t* socket = amqp_tcp_socket_new(conn);
    if (!socket) {
        qCWarning(connectionpool) << "创建TCP套接字失败";
        amqp_destroy_connection(conn);
        return nullptr;
    }
    
    // 连接到服务器
    int status = amqp_socket_open(socket, m_config.hostname.toUtf8().constData(), m_config.port);
    if (status != AMQP_STATUS_OK) {
        qCWarning(connectionpool) << "连接到服务器失败:" << amqp_error_string2(status);
        amqp_destroy_connection(conn);
        return nullptr;
    }
    
    // 登录
    amqp_rpc_reply_t reply = amqp_login(conn, 
                                       m_config.vhost.toUtf8().constData(),
                                       0,  // channel_max
                                       131072,  // frame_max (默认值)
                                       60,      // heartbeat (默认值)
                                       AMQP_SASL_METHOD_PLAIN,
                                       m_config.username.toUtf8().constData(),
                                       m_config.password.toUtf8().constData());
    
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        qCWarning(connectionpool) << "登录失败";
        amqp_destroy_connection(conn);
        return nullptr;
    }
    
    // 打开通道1（消费和发布消息都使用通道1）
    amqp_channel_open(conn, 1);
    reply = amqp_get_rpc_reply(conn);
    if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
        qCWarning(connectionpool) << "打开通道失败";
        amqp_destroy_connection(conn);
        return nullptr;
    }
    
    qCDebug(connectionpool) << "创建连接并打开通道成功";
    return conn;
}



/**
 * @brief 关闭连接
 * @param conn 要关闭的连接
 */
void MQConnectionPool::closeConnection(amqp_connection_state_t conn)
{
    if (conn) {
        try {
            amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
            amqp_destroy_connection(conn);
        } catch (...) {
            qCWarning(connectionpool) << "关闭连接时发生异常";
        }
    }
}

/**
 * @brief 查找连接信息
 * @param conn 连接状态指针
 * @return 连接信息指针，nullptr表示未找到
 */
std::shared_ptr<MQConnectionInfo> MQConnectionPool::findConnectionInfo(amqp_connection_state_t conn)
{
    for (auto& connInfo : m_connections) {
        if (connInfo->connection && connInfo->connection.get() == conn) {
            return connInfo;
        }
    }
    return nullptr;
}



/**
 * @brief 更新统计信息
 */
void MQConnectionPool::updateStatistics()
{
    //GJ todo  updateStatistics有问题
    //暂时不用
    return;

    QMutexLocker statsLocker(&m_statisticsMutex);
    
    m_statistics.totalConnections = m_connections.size();
    m_statistics.activeConnections = 0;
    m_statistics.idleConnections = 0;
    
    for (const auto& connInfo : m_connections) {
        if (connInfo->inUse) {
            m_statistics.activeConnections++;
        } else {
            m_statistics.idleConnections++;
        }
    }
    
    // 计算平均等待时间
    if (!m_waitTimes.isEmpty()) {
        qint64 totalWaitTime = 0;
        QDateTime now = QDateTime::currentDateTime();
        
        for (const QDateTime& waitTime : m_waitTimes) {
            totalWaitTime += waitTime.msecsTo(now);
        }
        
        m_statistics.averageWaitTime = static_cast<double>(totalWaitTime) / m_waitTimes.size();
    } else {
        m_statistics.averageWaitTime = 0.0;
    }
    
    emit statisticsUpdated(m_statistics);
}
